package com.atguigu.chapter07.d_state;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @ClassName: Flink01_State_Operator_List
 * @Description:
 * @Author: kele
 * @Date: 2021/4/7 19:38
 *
 *
 * 键控状态之 Aggregate
 *
 * 通过runtime获取
 *
 * 案例： 计算每个传感器的平均水位
 * 思路： 将数据存到元组中，第一个放数据，第二个放数量，第一个数据相加得数据和 除以 元素个数
 *
 * 1、先重写open方法，获取状态
 *
 * aggState = getRuntimeContext()
 * 		.getAggregatingState(new AggregatingStateDescriptor<Integer, Tuple2<Long, Long>, Double>(
 * 				"名称"
 * 				//处理方法
 * 				new AggregateFunction<Integer, Tuple2<Long, Long>, Double>() {
 * 					@Override
 * 					public Tuple2<Long, Long> createAccumulator() {
 * 						//初始化，将初始的变量赋值
 * 					}
 * 					@Override
 * 					public Tuple2<Long, Long> add(Integer value, Tuple2<Long, Long> acc) {
 * 						//计算方法 ，中间的计算过程稿
 * 					}
 * 					@Override
 * 					public Double getResult(Tuple2<Long, Long> acc) {
 * 						//返回结果，将中间计算的结果保存
 * 					}
 * 					@Override
 * 					public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
 * 						//只有在会话窗口的时候才会进行处理
 * 					}
 * 				},
 * 				//agg的类型
 * 		));
 * }
 *
 *
 * 2、将接收到的数据放入到状态中
 *     aggState.add(value);
 *
 **/
public class Flink06_Key_State_Aggregate {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        env.socketTextStream("hadoop162",8888)
                .map(datas->{

                    String[] data = datas.split(",");
                    return new WaterSensor(data[0],
                            Long.parseLong(data[1]) * 1000,
                            Integer.valueOf(data[2]));
                })
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    private AggregatingState<Integer, Double> aggState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        aggState = getRuntimeContext()
                                .getAggregatingState(new AggregatingStateDescriptor<Integer, Tuple2<Long, Long>, Double>(
                                        "AggState",

                                        //处理方法
                                        new AggregateFunction<Integer, Tuple2<Long, Long>, Double>() {

                                            //初始化
                                            @Override
                                            public Tuple2<Long, Long> createAccumulator() {
                                                return Tuple2.of(0l,0l);
                                            }

                                            //计算方法
                                            @Override
                                            public Tuple2<Long, Long> add(Integer value, Tuple2<Long, Long> acc) {
                                                return Tuple2.of(acc.f0+value,acc.f1+1);
                                            }

                                            //返回结果
                                            @Override
                                            public Double getResult(Tuple2<Long, Long> acc) {
                                                return acc.f0 * 1.0 / acc.f1;
                                            }

                                            //只有在会话窗口的时候才会进行处理
                                            @Override
                                            public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
                                                return Tuple2.of(a.f0+b.f0,a.f1+b.f1);
                                            }
                                        },
                                        //agg的类型
                                        Types.TUPLE(Types.LONG, Types.LONG)

                                ));
                    }

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                        //将数据放入到agg中
                        aggState.add(value.getVc());

                        out.collect(ctx.getCurrentKey()+"  平均值为："+aggState.get());


                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
